package io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler;

import io.kiki.sba.registry.common.model.dataserver.Datum;
import io.kiki.sba.registry.common.model.slot.SlotAccess;
import io.kiki.sba.registry.common.model.slot.SlotAccessGenericResponse;
import io.kiki.sba.registry.common.model.store.SubDatum;
import io.kiki.sba.registry.compress.CompressUtils;
import io.kiki.sba.registry.compress.Compressor;
import io.kiki.sba.registry.server.data_store.providedata.CompressDatumService;
import io.kiki.sba.registry.server.shared.util.DatumUtils;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

import static io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler.HandlerMetrics.GetData.*;


public abstract class BaseGetDataHandler<T> extends AbstractDataHandler<T> {

    private static final Logger logger = LoggerFactory.getLogger(BaseGetDataHandler.class);

    @Autowired
    protected ThreadPoolExecutor threadPoolExecutor;

    @Getter
    @Resource
    private CompressDatumService compressDatumService;

    @Override
    public Executor getExecutor() {
        return threadPoolExecutor;
    }

    protected SlotAccessGenericResponse<SubDatum> processSingleDataCenter(String dataCenter, String dataInfoId, long slotsEpoch, long slotLeaderEpoch, String[] acceptEncodes) {
        final SlotAccess slotAccess = checkSlotAccess(dataCenter, dataInfoId, slotsEpoch, slotLeaderEpoch);
        if (!slotAccess.isAccept()) {
            GET_DATUM_N_COUNTER.inc();
            return buildResponse(false, slotAccess, null, "slotAccess not accept.");
        }

        final Datum datum = datumStorageDelegate.get(dataCenter, dataInfoId);
        // important. double check the slot access. avoid the case:
        // 1. the slot is leader, the first check pass
        // 2. slot moved and data cleaned
        // 3. get datum, but null after cleaned, dangerous!!
        // 3.1. session get datum by change.version, ignored null datum, would not push
        // 3.2. session get datum by subscriber.register, accept null datum(the pub may not exists) and
        // push empty
        // so, need a double check slot access, make sure the slot's leader not change in the getting
        final SlotAccess slotAccessAfter = checkSlotAccess(dataCenter, dataInfoId, slotsEpoch, slotLeaderEpoch);
        if (slotAccessAfter.getSlotLeaderEpoch() != slotAccess.getSlotLeaderEpoch()) {
            // the slot's leader has change
            GET_DATUM_N_COUNTER.inc();
            return buildResponse(false, slotAccessAfter, null, "slotLeaderEpoch has change, prev=" + slotAccess);
        }

        GET_DATUM_Y_COUNTER.inc();

        SubDatum zipDatum = zipDatum(dataCenter, dataInfoId, datum, acceptEncodes);

        return buildResponse(true, slotAccessAfter, zipDatum, "");
    }

    private SubDatum zipDatum(String dataCenter, String dataInfoId, Datum datum, String[] acceptEncodes) {
        // return SubDatum, it's serdeSize and memoryOverhead much smaller than Datum
        SubDatum subDatum = datum != null ? DatumUtils.of(datum) : null;

        Compressor compressor = compressDatumService.getCompressor(subDatum, acceptEncodes);

        SubDatum zipDatum = DatumUtils.compressSubDatum(subDatum, compressor);

        String encode = "";
        if (compressor != null) {
            encode = compressor.getEncoding();
        }

        if (subDatum != null) {
            logger.info("getD,{},{},{},{},encode={},dataBoxSize={},encodeSize={}", dataInfoId, dataCenter, subDatum.getUnzippedSubPublisherList().size(), subDatum.getVersion(), CompressUtils.normalizeEncode(encode), zipDatum.getDataBoxSizeCache(), zipDatum.size());
            GET_PUBLISHER_COUNTER.inc(subDatum.getUnzippedSubPublisherList().size());
        } else {
            logger.info("getNilD,{},{}", dataInfoId, dataCenter);
        }

        return zipDatum;
    }

    private SlotAccessGenericResponse<SubDatum> buildResponse(boolean success, SlotAccess slotAccess, SubDatum subDatum, String msg) {
        return new SlotAccessGenericResponse<>(success, msg, slotAccess, subDatum);
    }


}
